package system.config.rocketmq.consumer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import common.config.rocketmq.MqTaskEnum;
import common.config.rocketmq.RocketMqConstants;
import common.service.CommonService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import system.model.request.UpdateOperatorRequest;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author 米泽鹏
 * @since 2022-5-11 下午 6:08
 */
@Component
@Slf4j
public class UpdateUsernameOrNicknameMqConsumer extends DefaultMQPushConsumer implements InitializingBean, DisposableBean {

	@Resource
	private CommonService commonService;
	@Value("${spring.application.name}")
	private String appName;

	@Override
	public void afterPropertiesSet() throws Exception {
		// 初始化时声明消费者组名称
		setConsumerGroup(MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getConsumerGroup());
		// 指明 nameserver 地址
		setNamesrvAddr(RocketMqConstants.NAMESERVER_ADDRESS);
		// 订阅一个或多个 topic 和 tag（多个tag用||隔开）
		subscribe(MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getTopic(), SubscriptionData.SUB_ALL);
		// 消息模型：集群模式
		setMessageModel(MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getMessageModel());
		// 注册消息到达时执行的回调方法
		registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
			try {
				for (MessageExt messageExt : messages) {
					String message = new String(messageExt.getBody());
					UpdateOperatorRequest request = JSONUtil.toBean(message, UpdateOperatorRequest.class);
					String oldOperator = request.getOldOperator();
					String newOperator = request.getNewOperator();
					Assert.isTrue(StrUtil.isNotBlank(oldOperator), "oldOperator不能为空");
					Assert.isTrue(StrUtil.isNotBlank(newOperator), "newOperator不能为空");
					// 更新各个表的operator字段
					List<String> tableNameList = commonService.getAllTableName("satisfactory").stream().filter(tableName -> tableName.startsWith("sys_") || tableName.startsWith("g_")).collect(Collectors.toList());
					for (String tableName : tableNameList) {
						commonService.updateOperator(tableName, oldOperator, newOperator);
					}
				}
				log.info("{}消息消费成功，消息数量：{}", MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getDescription(), messages.size());
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			} catch (Exception e) {
				log.error("{}消息消费发生异常", MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getDescription(), e);
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}
		});
		// 消费失败重试次数
		setMaxReconsumeTimes(3);
		// 启动消费者
		start();
		log.info("RocketMQ Consumer启动成功，分组名称：{}，来自服务：{}", MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getConsumerGroup(), appName);
	}

	@Override
	public void destroy() {
		shutdown();
		log.info("RocketMQ Consumer关闭，分组名称：{}，来自服务：{}", MqTaskEnum.UPDATE_USERNAME_OR_NICKNAME.getConsumerGroup(), appName);
	}

}
